package cn.dmp.tools

import cn.dmp.beans.Log
import cn.dmp.utils.{NumberFormat, SchemaUtils}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext, SaveMode}

/**
  *
  * 将原始文件日志转换成parquet文件格式
  * 采用snappy压缩格式
  *   第二种方式：使用自定义类的方式构建schema信息。
  */

object Bizp2Parquet_Version2 {
  def main(args: Array[String]): Unit = {
    // 0.检验参数的个数。最低需要两个：日志输入目录，日志输出目录。
    if (args.length != 2) {
      println(
        """
          |cn.dmp.tools.Bzip2Parquet
          |参数：
          | logInputPath
          | resultOutputPath
        """.stripMargin)
      sys.exit()
    }

    // 1.接收程序的参数。
    val Array(logInputPath, resultOutputPath) = args

    // 2.创建sparkconf->sparkContext。//s"${this.getClass.getSimpleName}"得到大明湖起哦按类的名称。
    val conf = new SparkConf().setAppName(s"${this.getClass.getSimpleName}").setMaster("local[*]")
    // RDD序列化到磁盘，worker与worker之间的传输。
    /**
      * spark优化：spark.serializer修改序列化方式
      * 进行节点的数据传递，或者保存数据时都会进行序列化。
      * spark默认的是org.apache.spark.serializer.JavaSerializer。
      * 而我们要修改成org.apache.spark.serializer.KryoSerializer。
      */
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    //输出数据的时候的压缩编码设置为snappy,spark2.0默认就是snappy。可以为<snappy,gzip,lzo>
    conf.set("spark.sql.parquet.compression.codec", "snappy")

    /**
      *
      * 因为我们是自定义类Log,因此必须将自定义类注册一下：即注册自定义类的序列化方式
      */
    conf.registerKryoClasses(Array(classOf[Log]))

    val sc = new SparkContext(conf)

    val sQLContext = new SQLContext(sc)


    // 3.读取日志数据
    val rawdata: RDD[String] = sc.textFile(logInputPath)

    // 4.根据业务需求对数据进行ETL:ETL是将业务系统的数据经过抽取、清洗转换之后加载到数据仓库的过程
    //xxx,x,x,x,,,,比如有这样的8个数据，用split(",")来切的话，后面这几个,,,的获取不到。必须用line.split(",",line.length或者-1)代表一直切分到这一行的最后。
    // 这里-1
    /**
      * 接下来，构建scheme信息，这次就不是通过StructType来构建了，而是通过一个自定义类来构建的。
      * 创建一个类Log类  继承Product类，并且实现with接口Serializable
      */
    //注意自定义类的返回数据的格式  RDD[Log] ，里面是Log
    val dataLog: RDD[Log] = rawdata
      .map(line => line.split(",", -1))
      .filter(_.length >= 85)
      .map(arr => Log(arr))


    // 5.将结果存储到本地磁盘!
    val dataFrame: DataFrame = sQLContext.createDataFrame(dataLog)

    /**
      * 在将数据存储为parquet文件之前。我们还可以进行分区。分区后在parquet
      * 如下：按照"Provincename", "Cityname"来进行分区。
      */
    dataFrame.write.mode(SaveMode.Overwrite).partitionBy("Provincename", "Cityname").parquet(resultOutputPath)

    // 6.关闭sc
    sc.stop()

    //本地测试运行：!!!!!注意提前设置好参数：输入目录和输出目录
    /**
      * 输入目录：E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\data\2016-10-01_06_p1_invalid.1475274123982.log.FINISH
      * 输出目录: E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\biz2parquet_Version2
      *
      */

  }
}
